package com.dl.mqtest.controller;

import jakarta.annotation.Resource;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.Semaphore;

/**
 * @author DL
 * @create 2023-09-11 15:43
 */
@RestController
@RequestMapping("/rocketmq")
public class RocketMqController {

    @Resource
    private RocketMQTemplate rocketMqTemplate;


    @RequestMapping("/syncSend/{value}")
    public String syncProducer(@PathVariable String value){
        rocketMqTemplate.syncSend("test-topic",value);
        return "同步步消息发送成功";
    }

    @RequestMapping("/asyncSend")
    public String asyncProducer(){

        for (int i = 0; i < 100; i++) {
            int finalI = i;
            rocketMqTemplate.asyncSend("test-topic",i + "-异步消息", new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println(finalI + "-异步消息发送成功" + sendResult);
                }

                @Override
                public void onException(Throwable throwable) {
                    System.out.println(finalI + "-异步消息发送失败" + throwable.getMessage());
                }
            });
        }


        return "success";
    }


    @RequestMapping("/orderSend")
    public String orderProducer() throws InterruptedException {
        Semaphore semaphore = new Semaphore(5); // 限制最多10个并发

        for (int i = 0; i < 100; i++) {
            int finalI = i;
            // 获取一个许可证。如果当前没有可用的许可证，则调用会被阻塞，直到有许可证可用
            semaphore.acquire();
            rocketMqTemplate.asyncSendOrderly("test-topic",i + "-顺序异步消息", "order", new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    // 释放一个许可证。这会增加可用许可证的数量，从而允许其他被阻塞的 acquire() 调用继续执行。
                    semaphore.release();
                    System.out.println(finalI + "-顺序异步消息发送成功" + sendResult);
                }

                @Override
                public void onException(Throwable throwable) {
                    semaphore.release();
                    System.out.println(finalI + "-顺序异步消息发送失败" + throwable.getMessage());
                }
            });
        }


        return "success";
    }



}
